/*
 * Copyright (c) 2022 The red-star Project
 *
 * Licensed under the Apache License, version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.inyourcode.core.monitor;

import com.inyourcode.core.GlobalConstants;
import com.inyourcode.core.transport.netty.NettyTcpAcceptor;
import com.inyourcode.core.transport.netty.TcpChannelProvider;
import com.inyourcode.core.util.StackTraceUtil;
import com.inyourcode.core.util.Strings;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.List;

/**
 * @author JackLei
 */
public class ConsoleAcceptor extends NettyTcpAcceptor {
    private static final Logger logger = LoggerFactory.getLogger(ConsoleAcceptor.class);
    private static final int DEFAULT_PORT = 20000;

    // handlers
    private final TelnetHandler handler = new TelnetHandler();
    private final StringEncoder encoder = new StringEncoder(GlobalConstants.UTF8);

    public ConsoleAcceptor() {
        this(DEFAULT_PORT);
    }

    public ConsoleAcceptor(int port) {
        super(port, 1, false);
    }

    @Override
    public ChannelFuture bind(SocketAddress localAddress) {
        ServerBootstrap boot = bootstrap();
        if (isNativeEt()) {
            boot.channelFactory(TcpChannelProvider.NATIVE_ACCEPTOR);
        } else {
            boot.channelFactory(TcpChannelProvider.NIO_ACCEPTOR);
        }

        boot.childHandler(new ChannelInitializer<Channel>() {

            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(
                        new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()),
                        new StringDecoder(GlobalConstants.UTF8),
                        encoder,
                        new LineEncoder(),
                        handler);
            }
        });

        setOptions();

        return boot.bind(localAddress);
    }

    @Override
    public void start() throws InterruptedException {
        super.start(false);
    }

    @ChannelHandler.Sharable
    class TelnetHandler extends ChannelInboundHandlerAdapter {

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            Channel ch = ctx.channel();

            if (msg instanceof String) {
                String[] args = Strings.split(((String) msg).replace("\r\n", ""), ' ');
                if (args == null || args.length == 0) {
                    return;
                }
                MonitorService.getInstance().processCMD(ch, args);
            } else {
                logger.warn("Unexpected message type received: {}, channel: {}.", msg.getClass(), ch);
                ReferenceCountUtil.release(msg);
            }
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(GlobalConstants.NEWLINE + "Welcome to red star monitor! Please auth with password." + GlobalConstants.NEWLINE);
            MonitorService.getInstance().getMonitorOpt("help").process(ctx.channel(), null);
        }


        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            logger.error("An exception was caught: {}, channel {}.", StackTraceUtil.stackTrace(cause), ctx.channel());
            ctx.close();
        }
    }

    static class LineEncoder extends MessageToMessageEncoder<String> {

        @Override
        protected void encode(ChannelHandlerContext ctx, String msg, List<Object> out) throws Exception {
            out.add(GlobalConstants.NEWLINE + msg + GlobalConstants.NEWLINE);
        }
    }
}
